package com.accenture.automation.biz.consumer;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.env.Environment;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import com.accenture.automation.biz.service.InterfaceRequestService;
import com.accenture.automation.common.CommonConstants;
import com.accenture.automation.common.SpringContextUtil;

@Component
public class ConsumerListener {
	protected final Logger logger = LoggerFactory.getLogger(this.getClass());

	final String topic = "TEST";
	@Autowired
	private Environment env;

	@KafkaListener(topics = { topic })
	public void listen(ConsumerRecord<?, ?> record) {
		handler(record);
		((Consumer<?, ?>) SpringContextUtil.getBean("Consumer")).commitSync();
		System.out.println("********************************消费kafka的offset****************************************:" + record.offset());

	}

	/**
	 * 数据处理
	 * 
	 * @param record
	 */
	private void handler(ConsumerRecord<?, ?> record) {
		// kafka的key创捷格式:accountPostApp_accountInid_accountPK
		String[] records = record.key().toString().split("_");
		// 根据key获取配置文件中的接口地址
		String url = env.getProperty(records[0] + "_" + records[1]);
		String message = record.value().toString();
		// 获取接口重试次数
		int retryCount = Integer.parseInt(env.getProperty("retryCount").toString());
		// 获取回写ITE开关
		int callBackITE = Integer.parseInt(env.getProperty("callBackITE").toString());
		// 根据kafka key 发送ITE
		if (records[0].equals(CommonConstants.ROOT_APP)) {
			InterfaceRequestService.postITE(url, message, retryCount);
		}
		// 根据kafka key 发送DOE
		else if (records[0].equals(CommonConstants.ACCOUNT_DOE)) {
			// 上抛到对方系统
			String resultStr = InterfaceRequestService.accountDoePost(url, message, retryCount,
					env.getProperty("DOE_TOKEN").toString());
			if (callBackITE == 0) {
				callBackITE(resultStr, records[2], retryCount);
			}

		} else {
			// 上抛到对方系统
			String resultStr = InterfaceRequestService.accountPost(url, message, retryCount);
			if (callBackITE == 0) {
				callBackITE(resultStr, records[2], retryCount);
			}

		}
	}

	/**
	 * 回写上抛结果到ITE
	 * 
	 * @param resultStr
	 * @param accountPk
	 * @param retryCount
	 */
	private void callBackITE(String resultStr, String accountPk, int retryCount) {
		// 格式化回写ITE地址
		String callBackUrl = env.getProperty("ITE_ACCOUNT_STATUS");
		callBackUrl = InterfaceRequestService.urlFormat(callBackUrl, accountPk);
		// 上抛结果回写ITE
		InterfaceRequestService.postITE(callBackUrl, resultStr, retryCount);
	}

}